package com.chb.storm.kafkatopology;

import java.util.UUID;

import storm.kafka.KafkaSpout;
import storm.kafka.SpoutConfig;
import storm.kafka.StringScheme;
import storm.kafka.ZkHosts;
import backtype.storm.Config;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.AuthorizationException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.spout.SchemeAsMultiScheme;
import backtype.storm.topology.TopologyBuilder;

public class KafkaStreamPrint {

	public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException, AuthorizationException {
		// TODO Auto-generated method stub
		//kafka的zookeeper配置
//		ZkHosts zkHosts = new ZkHosts("njtest03:2181");
		//kafka的zookeeper根目录+"/brokers"
		ZkHosts zkHosts = new ZkHosts("njtest03:2181", "/kafka_zk/brokers");
		//topic的name
		String topicName = "mytopic";
		//kafka的zookeeper根目录
//		String zkRoot = "/kafka_zk";
		String zkRoot = "/kafka_zk";
		//The id should uniquely identify your spout
		String spoutId = "kafkaStreamPrintId=" + UUID.randomUUID().toString();
		
		SpoutConfig kafkaConfig = new SpoutConfig(zkHosts, topicName, zkRoot, spoutId);
		
		//Specify that the kafka messages are String
		kafkaConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
//		kafkaConfig.forceFromStart = true;
		
		TopologyBuilder builder = new TopologyBuilder();
		builder.setSpout("KafkaSpout", new KafkaSpout(kafkaConfig), 1);
		builder.setBolt("PrinterBolt", new PrinterBolt()).globalGrouping("KafkaSpout");
		
		Config config = new Config();
		StormSubmitter.submitTopology("KafkaStreamPrint", config, builder.createTopology());
	}

}
